package com.surfilter.massdata.spark.task.daystati;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.storage.StorageLevel;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.ActiveIpBean;
import com.surfilter.massdata.spark.bean.SAN051;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.IPDatabase;

public class ActiveIPAnalyzTask_New implements ISparkTask {
	private static final long serialVersionUID = 1L;
	private String city_top;
	private String oper_top;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		try{
			DataFrame cipFilterFrame = dataFrames.get("dwb_domain_cip");
			SQLContext sqlContext = cipFilterFrame.sqlContext();
			
			DataFrame operators_id_param=sqlContext.sql("select distinct OPERATORS_ID as operators_id from br2002_temp");
			operators_id_param.registerTempTable("operators_id_param");
			
			DataFrame city_param = sqlContext.sql("select distinct CITY as city from br2002_temp");
			city_param.registerTempTable("city_param");
	
			String dayStr = commandMap.get("-d");
			Date date = DateUtil.getExecDate(new Date(), dayStr);
			// 活跃IP排名-地市用户IP访问量排名（50）
			DataFrame city_top = getCityTopDF(city_param, date);
			list.add(new OutQueueEntity(this.city_top, city_top));

			// 活跃IP排名-运营商用户IP访问量排名（50）
			DataFrame oper_top = getOperatorTopDF(operators_id_param, date);
			list.add(new OutQueueEntity(this.oper_top, oper_top));
			
			//System.out.println("city_top:"+city_top.count());
			//System.out.println("oper_top:"+oper_top.count());
			
			sqlContext.dropTempTable("operators_id_param");
			sqlContext.dropTempTable("city_param");
			//CommonUtils.deleteTaskTableData("SAN051", date, "WEBSITE_COUNT_AREA", 0, "day");
			//CommonUtils.deleteTaskTableData("SAN051", date, "WEBSITE_COUNT_OPERATORS", 0, "day");
		}
		catch(Exception e){
			SysLog.error(e.getMessage());
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("ActiveIPAnalyzTask_New:exectime: "+min+" min............");
		
		return list;
	}

	@SuppressWarnings("serial")
	private DataFrame getOperatorTopDF(DataFrame operators_id_param, Date date) {
		String sql = "select operators_id,cip,sum(visit_count) as count_value from dwb_domain_cip_temp where operators_id is not null group by operators_id,cip order by operators_id,count_value desc";
		DataFrame oper_tmp = operators_id_param.sqlContext().sql(sql);
		JavaRDD<Row> operRDD = oper_tmp.toJavaRDD();
		operRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());
		
		List<SAN051> list = new ArrayList<SAN051>();
		List<Row> rows = operators_id_param.toJavaRDD().collect();
		for(int i = 0 ; i < rows.size() ; i++){
			final String my_operators_id = rows.get(i).getAs("operators_id");
			
			List<Row> operIpList = operRDD.filter(new Function<Row, Boolean>() {

				@Override
				public Boolean call(Row row) throws Exception {
					String operators_id = row.getAs("operators_id");
					if(StringUtils.equals(my_operators_id, operators_id)){
						return true;
					}
					return false;
				}
			}).take(50);
			
			if (operIpList.size() > 0) {
				for (int j = 0; j < operIpList.size(); j++) {
					SAN051 sa = new SAN051();
					Row trow = operIpList.get(j);
					sa.setYear(DateUtil.getCurrentYear(date));
					sa.setHalf_year(DateUtil.getHalfYear(date));
					sa.setQuarter(DateUtil.getQuarter(date));
					sa.setMonth(DateUtil.getCurrentMonth(date));
					sa.setWeek(DateUtil.getCurrentWeek(date));
					sa.setDay(DateUtil.getCurrentDay(date));
					sa.setHour(0);
					sa.setBuss_type("WEBSITE_COUNT_OPERATORS");
					sa.setBuss_value(my_operators_id);
					sa.setCount_value(Long.parseLong(trow.getAs("count_value").toString()));
					sa.setSta_range(0);
					sa.setClient_ip(trow.getAs("cip").toString());
					sa.setRank(j + 1);
					list.add(sa);
				}
			}
		}
		operRDD.unpersist();
		
		// 生成DataFrame
		JavaSparkContext sc = JavaSparkContext.fromSparkContext(operators_id_param.sqlContext().sparkContext());
		JavaRDD<SAN051> city_result_rdd = sc.parallelize(list);
		DataFrame oper_top = operators_id_param.sqlContext().createDataFrame(city_result_rdd, SAN051.class);
		return oper_top;
	}

	@SuppressWarnings("serial")
	private DataFrame getCityTopDF(DataFrame city_param, Date date) {
		String sql = "select city,cip,sum(visit_count) as count_value from dwb_domain_cip_temp where city is not null group by city,cip order by city,count_value desc";
		DataFrame city_tmp = city_param.sqlContext().sql(sql);
		JavaRDD<Row> cityRDD = city_tmp.toJavaRDD();
		cityRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());
		
		List<Row> citys = city_param.toJavaRDD().collect();
		// 生成结果集
		List<SAN051> list = new ArrayList<SAN051>();
		for (int i = 0; i < citys.size(); i++) {
			Row row = citys.get(i);
			final String myCity = row.getString(0);
			List<Row> cityIpList = cityRDD.filter(new Function<Row, Boolean>() {

				@Override
				public Boolean call(Row row) throws Exception {
					String city = row.getAs("city");
					if(StringUtils.equals(myCity, city)){
						return true;
					}
					return false;
				}
			}).take(50);
			
			for (int j = 0; j < cityIpList.size(); j++) {
				SAN051 sa = new SAN051();
				Row trow = cityIpList.get(j);
				sa.setYear(DateUtil.getCurrentYear(date));
				sa.setHalf_year(DateUtil.getHalfYear(date));
				sa.setQuarter(DateUtil.getQuarter(date));
				sa.setMonth(DateUtil.getCurrentMonth(date));
				sa.setWeek(DateUtil.getCurrentWeek(date));
				sa.setDay(DateUtil.getCurrentDay(date));
				sa.setHour(0);
				sa.setBuss_type("WEBSITE_COUNT_AREA");
				sa.setBuss_value(myCity);
				sa.setCount_value(Long.parseLong(trow.getAs("count_value").toString()));
				sa.setSta_range(0);
				sa.setClient_ip(trow.getAs("cip").toString());
				sa.setRank(j + 1);
				list.add(sa);
			}
		}
		cityRDD.unpersist();
		// 生成DataFrame
		JavaSparkContext sc = JavaSparkContext.fromSparkContext(city_param.sqlContext().sparkContext());
		JavaRDD<SAN051> city_result_rdd = sc.parallelize(list);
		DataFrame city_top = city_param.sqlContext().createDataFrame(city_result_rdd, SAN051.class);
		return city_top;
	}

	private DataFrame testNADF(SQLContext sqlContext,
			DataFrame operators_id_param) {
		String na = "1";
		String sql = "select * from operators_id_param where OPERATORS_ID= \'"
				+ na + "\' ";
		DataFrame oper_tmp = sqlContext.sql(sql);
		return oper_tmp;
	}
}
